package drds.binlog.sink.entry;

import drds.binlog.common.position.LogIdentity;
import drds.binlog.protocol.entry_protocol_proto.Entry;
import drds.binlog.protocol.entry_protocol_proto.EntryType;
import drds.binlog.sink.AbstractEventSink;
import drds.binlog.sink.EventDownStreamHandler;
import drds.binlog.sink.EventSink;
import drds.binlog.sink.exception.BinlogSinkException;
import drds.binlog.store.EventStore;
import drds.binlog.store.model.Event;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;

/**
 * mysql binlog数据对象输出
 */
public class EntryEventSink extends AbstractEventSink<List<Entry>> implements EventSink<List<Entry>>
{

    private static final Logger logger = LoggerFactory.getLogger(EntryEventSink.class);
    private static final int maxFullTimes = 10;
    protected boolean filterTransactionEntry = false;                                        // 是否需要过滤事务头/尾
    protected boolean filterEmtryTransactionEntry = true;                                         // 是否需要过滤空的事务头/尾
    protected long emptyTransactionInterval = 5 * 1000;                                     // 空的事务输出的频率
    protected long emptyTransctionThresold = 8192;                                         // 超过1024个事务头，输出一个
    protected volatile long lastEmptyTransactionTimestamp = 0L;
    protected AtomicLong lastEmptyTransactionCount = new AtomicLong(0L);
    private EventStore<Event> eventStore;
    private AtomicLong eventsSinkBlockingTime = new AtomicLong(0L);

    public EntryEventSink()
    {
        addHandler(new HeartBeatEntryEventHandler());
    }

    public void start()
    {
        super.start();
        Assert.notNull(eventStore);

        for (EventDownStreamHandler eventDownStreamHandler : getHandlers())
        {
            if (!eventDownStreamHandler.isRunning())
            {
                eventDownStreamHandler.start();
            }
        }
    }

    public void stop()
    {
        super.stop();

        for (EventDownStreamHandler eventDownStreamHandler : getHandlers())
        {
            if (eventDownStreamHandler.isRunning())
            {
                eventDownStreamHandler.stop();
            }
        }
    }


    public boolean sink(List<Entry> entryList, InetSocketAddress remoteAddress, String destination)
            throws BinlogSinkException,
            InterruptedException
    {
        List rowDatas = entryList;
        if (filterTransactionEntry)
        {
            rowDatas = new ArrayList<Entry>();
            for (Entry entry : entryList)
            {
                if (entry.getEntryType() == EntryType.ROWDATA)
                {
                    rowDatas.add(entry);
                }
            }
        }

        return sinkData(rowDatas, remoteAddress);
    }

    private boolean sinkData(List<Entry> entryList, InetSocketAddress remoteAddress)
            throws InterruptedException
    {
        boolean hasRowData = false;
        boolean hasHeartBeat = false;
        List<Event> events = new ArrayList<Event>();
        for (Entry entry : entryList)
        {
            if (!doFilter(entry))
            {
                continue;
            }

            Event event = new Event(new LogIdentity(remoteAddress, -1L), entry);
            events.add(event);
            hasRowData |= (entry.getEntryType() == EntryType.ROWDATA);
            hasHeartBeat |= (entry.getEntryType() == EntryType.HEARTBEAT);
        }

        if (hasRowData)
        {
            // 存在row记录
            return doSink(events);
        } else if (hasHeartBeat)
        {
            // 存在heartbeat记录，直接跳给后续处理
            return doSink(events);
        } else
        {
            // 需要过滤的数据
            if (filterEmtryTransactionEntry && !CollectionUtils.isEmpty(events))
            {
                long currentTimestamp = events.get(0).getExecuteDateTimeLong();
                // 基于一定的策略控制，放过空的事务头和尾，便于及时更新数据库位点，表明工作正常
                if (Math.abs(currentTimestamp - lastEmptyTransactionTimestamp) > emptyTransactionInterval
                        || lastEmptyTransactionCount.incrementAndGet() > emptyTransctionThresold)
                {
                    lastEmptyTransactionCount.set(0L);
                    lastEmptyTransactionTimestamp = currentTimestamp;
                    return doSink(events);
                }
            }

            // 直接返回true，忽略空的事务头和尾
            return true;
        }
    }

    protected boolean doFilter(Entry entry)
    {
        if (eventFilter != null && entry.getEntryType() == EntryType.ROWDATA)
        {
            String name = getSchemaNameAndTableName(entry);
            boolean need = eventFilter.filter(name);
            if (!need)
            {
                logger.debug("eventFilter name[{}] entry : {}:{}",
                        name,
                        entry.getHeader().getLogfileName(),
                        entry.getHeader().getLogfileOffset());
            }

            return need;
        } else
        {
            return true;
        }
    }

    protected boolean doSink(List<Event> eventList)
    {
        for (EventDownStreamHandler<List<Event>> handler : getHandlers())
        {
            eventList = handler.before(eventList);
        }
        long blockingStart = 0L;
        int fullTimes = 0;
        do
        {
            if (eventStore.tryPut(eventList))
            {
                if (fullTimes > 0)
                {
                    eventsSinkBlockingTime.addAndGet(System.nanoTime() - blockingStart);
                }
                for (EventDownStreamHandler<List<Event>> handler : getHandlers())
                {
                    eventList = handler.after(eventList);
                }
                return true;
            } else
            {
                if (fullTimes == 0)
                {
                    blockingStart = System.nanoTime();
                }
                applyWait(++fullTimes);
                if (fullTimes % 100 == 0)
                {
                    long nextStart = System.nanoTime();
                    eventsSinkBlockingTime.addAndGet(nextStart - blockingStart);
                    blockingStart = nextStart;
                }
            }

            for (EventDownStreamHandler<List<Event>> handler : getHandlers())
            {
                eventList = handler.retry(eventList);
            }

        } while (running && !Thread.interrupted());
        return false;
    }

    // 处理无数据的情况，避免空循环挂死
    private void applyWait(int fullTimes)
    {
        int newFullTimes = fullTimes > maxFullTimes ? maxFullTimes : fullTimes;
        if (fullTimes <= 3)
        { // 3次以内
            Thread.yield();
        } else
        { // 超过3次，最多只sleep 10ms
            LockSupport.parkNanos(1000 * 1000L * newFullTimes);
        }

    }

    private String getSchemaNameAndTableName(Entry entry)
    {
        return entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName();
    }

    public void setEventStore(EventStore<Event> eventStore)
    {
        this.eventStore = eventStore;
    }

    public void setFilterTransactionEntry(boolean filterTransactionEntry)
    {
        this.filterTransactionEntry = filterTransactionEntry;
    }

    public void setFilterEmtryTransactionEntry(boolean filterEmtryTransactionEntry)
    {
        this.filterEmtryTransactionEntry = filterEmtryTransactionEntry;
    }

    public void setEmptyTransactionInterval(long emptyTransactionInterval)
    {
        this.emptyTransactionInterval = emptyTransactionInterval;
    }

    public void setEmptyTransctionThresold(long emptyTransctionThresold)
    {
        this.emptyTransctionThresold = emptyTransctionThresold;
    }

    public AtomicLong getEventsSinkBlockingTime()
    {
        return eventsSinkBlockingTime;
    }

}
